Problem

The Snowflake connector previously resolved external stage lineage by generating a single dataset URN directly from the raw stage URL (e.g. s3://my-bucket/data/ → one S3 URN). This produces incorrect lineage because:

1. A stage typically points to a path prefix, not a single dataset — the actual datasets ingested by the S3/GCS/ABS source are individual folders/tables under that prefix.
2. The generated URN will never match URNs produced by the data lake source when that source was ingested with a platform_instance, breaking the lineage graph.

Solution

Graph-based stage resolution (opt-in)

When resolve_external_stage_lineage_via_graph is enabled, the connector looks up every existing dataset URN in DataHub whose path is rooted at the stage's URL path. A single stage can therefore fan out to multiple upstream datasets that the corresponding data lake source previously ingested.

source:
  type: snowflake
  config:
    # ...
    resolve_external_stage_lineage_via_graph: true
    # Optional: set if the S3/GCS/ABS source was ingested with a platform_instance
    external_stage_platform_instance: "prod"

Changes

• New utility data_lake_common/path_urn_resolver.py — DataLakePathResolver bulk-fetches every dataset URN under a bucket once per (platform, platform_instance, bucket) and matches client-side. N stage paths sharing a bucket cost a single graph call, not N. Designed to be reusable across connectors (Redshift UNLOAD, external tables, etc.) and intentionally report-free; transient failures are surfaced via DataLakeUrnLookup.transient_error for callers to log and count.
• SnowflakeStagesExtractor — added _parse_external_stage_url, _compute_external_stage_urns, and _resolve_stage_via_graph. URL resolutions are cached per stage URL within a run; transient graph failures are not cached so a later stage with the same URL can retry.
• StageLookupEntry.dataset_urn: Optional[str] → dataset_urns: Tuple[str, ...] — stages can now resolve to multiple upstream datasets. The dataclass is frozen and constructed once with its final URNs.
• ParsedStageUrl — frozen dataclass; bucket is derived from path via a @property so the invariant holds by construction.
• snowflake_pipes.py — pipe extractor fans out every resolved stage URN into inputDatasets and deduplicates across stages using dict.fromkeys (insertion-order preserving, O(n)).
• snowflake_config.py — two new fields: resolve_external_stage_lineage_via_graph (default False) and external_stage_platform_instance (default None). Validators normalize blank platform instances to None and warn when external_stage_platform_instance is set without the resolve flag.
• snowflake_report.py — added external_stage_lineage_resolved and external_stage_lineage_unresolved counters.
• snowflake_v2.py — passes ctx.graph to SnowflakeStagesExtractor.
• Robustness — _compute_external_stage_urns now also catches InvalidUrnError so a malformed gcs:// (empty path is rejected by the URN constructor) no longer crashes extraction; the existing structured warning is emitted instead.
• Integration test fixtures — fixed naive datetime objects to use tzinfo=timezone.utc so containerized CI in non-UTC zones is deterministic.

False-positive prevention

A START_WITH URN filter for s3://my-bucket/folder would incorrectly match s3://my-bucket/folder_other. The dataset_path_is_rooted_at helper (in path_urn_resolver.py) rejects such false positives by requiring the character immediately after the prefix to be / (child path) or , (URN env separator, i.e. exact match). The same check also drops sibling buckets that the case-insensitive START_WITH wildcard might over-match.

Behaviour without the flag

When resolve_external_stage_lineage_via_graph is False (default), the connector falls back to the existing path-based URN generation — no change in behaviour for existing users.

Test plan

• tests/unit/data_lake_common/test_path_urn_resolver.py — covers the bucket-scoped fetch contract (one fetch per bucket, separate buckets fetched separately, scope is the bucket not the full path), sibling-bucket exclusion, transient-error pass-through with no caching, and the dataset_path_is_rooted_at boundary check parametrized across child/exact/sibling/wrong-platform/malformed inputs.
• TestGraphResolvedExternalStageLineage — S3/GCS/ABS resolution, empty matches, disabled-flag fallback, graph errors (with and without cache poisoning), platform-instance forwarding, URL caching, unsupported schemes, empty paths (both s3:// and gcs://), internal stages, and malformed URLs.
• TestExternalStageConfigValidators — blank/whitespace normalization, stripping, and the cross-field warning.
• TestSnowflakePipesExtractor — fan-out across multiple resolved URNs, cross-stage deduplication, and the stages_without_urn warning path (resolved/empty mix and the all-empty case).
• ./gradlew :metadata-ingestion:lint — ruff + mypy clean.
• ./gradlew :metadata-ingestion:testQuick — 546 Snowflake unit tests pass.

🤖 Generated with Claude Code